package com.feidee.fd.sml.algorithm.component.preprocess

import com.feidee.fdspark.transformer.{ColSelector, DateParser, MetaStorage, ModelType}
import org.apache.spark.ml.{Pipeline, PipelineModel, PipelineStage}
import org.apache.spark.sql.DataFrame

import scala.collection.mutable.ArrayBuffer

/**
  * @Author songhaicheng
  * @Date 2019/6/18 19:30
  * @Description 解析目标列日期信息预处理组件
  * @Reviewer
  */
case class ParseDateParam(
                           override val input_pt: String,
                           override val output_pt: String,
                           override val hive_table: String,
                           override val flow_time: String,
                           override val inputCol: String,
                           override val outputCol: String,
                           override val modelPath: String,
                           override val preserveCols: String,
                           format: String,
                           hourOfDayCol: String,
                           dayOfWeekCol: String,
                           dayOfMonthCol: String,
                           monthOfYearCol: String) extends PreprocessorParam {

  def this() = this(null, null, null, null, "input", "output", null, null, "yyyy-MM-dd HH:mm:ss", null, null, null, null)

  override def verify(): Unit = {
    super.verify()
    require(tool.isNotNull(format), "param format can't be null")
  }

  override def toMap: Map[String, Any] = {
    var map = super.toMap
    map += ("format" -> format)
    map += ("hourOfDayCol" -> hourOfDayCol)
    map += ("dayOfWeekCol" -> dayOfWeekCol)
    map += ("dayOfMonthCol" -> dayOfMonthCol)
    map += ("monthOfYearCol" -> monthOfYearCol)
    map
  }
}


class ParseDate extends AbstractPreprocessor[ParseDateParam] {
  override def train(param: ParseDateParam, data: DataFrame): PipelineModel = {
    // 把 preserveCols 取出来
    val preserveCols = if (tool.isNotNull(param.preserveCols)) {
      param.preserveCols.split(",")
    } else {
      // 不填 preserveCols 时，赋上一个空数组，不然上面会返回一个长度为 1 内容空字符串的数组！
      new Array[String](0)
    }

    // 结果列
    val arrayBuffer = ArrayBuffer[String]()
    if (tool.isNotNull(param.hourOfDayCol)){
      arrayBuffer += param.hourOfDayCol
    }
    if (tool.isNotNull(param.dayOfWeekCol)){
      arrayBuffer+=param.dayOfWeekCol
    }
    if (tool.isNotNull(param.dayOfMonthCol)){
      arrayBuffer+=param.dayOfMonthCol
    }
    if(tool.isNotNull(param.monthOfYearCol)){
      arrayBuffer+=param.monthOfYearCol
    }
    val cols = preserveCols ++ arrayBuffer


    // 存储训练参数元信息
    val meta = new MetaStorage()
      .setModelType(ModelType.Preprocessing)
      .setParameters(param.toMap)
      .setFields(data.schema.fieldNames)

    // 最终结果只保留结果列
    val colSelector = new ColSelector()
      .setCols(cols)

    // 执行流程：存储元信息 》 特征组件处理过程 》 保留结果列
    val pipeline = new Pipeline().setStages(Array(meta) ++ setUp(param, data) ++ Array(colSelector))
    val model = pipeline.fit(data)

    model
  }
  override def setUp(param: ParseDateParam, data: DataFrame): Array[PipelineStage] = {
    val parser = new DateParser()
      .setInputCol(param.inputCol)
      .setFormat(param.format)


    // 全部设置的情况下，默认全部生成，放到对应列下
    if (!tool.isNotNull(param.dayOfWeekCol) &&
      !tool.isNotNull(param.dayOfMonthCol) &&
      !tool.isNotNull(param.monthOfYearCol) &&
      !tool.isNotNull(param.hourOfDayCol)) {
      val fieldNames = data.schema.fieldNames

      parser.setDayOfWeekCol(tool.renameDuplicatedColName(s"${param.inputCol}__dayOfWeek", fieldNames))
      parser.setDayOfMonthCol(tool.renameDuplicatedColName(s"${param.inputCol}_dayOfMonth", fieldNames))
      parser.setMonthOfYear(tool.renameDuplicatedColName(s"${param.inputCol}_monthOfYear", fieldNames))
      parser.setHourOfDayCol(tool.renameDuplicatedColName(s"${param.inputCol}_hourOfDay", fieldNames))
    } else {
      if (param.dayOfWeekCol!=null){
        parser.setDayOfWeekCol(param.dayOfWeekCol)
      }
      if (param.dayOfMonthCol!=null){
        parser.setDayOfMonthCol(param.dayOfMonthCol)
      }
      if (param.monthOfYearCol!=null){
        parser.setMonthOfYear(param.monthOfYearCol)
      }

      if (param.hourOfDayCol!=null){
        parser.setHourOfDayCol(param.hourOfDayCol)
      }

    }

    Array(parser)
  }

}

object ParseDate {

  def apply(paramStr: String): Unit = {
    new ParseDate()(paramStr)
  }

  def main(args: Array[String]): Unit = {
    ParseDate(args(0))
  }
}




